package org.example.watemark;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

import java.time.Duration;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/19 15:53
 */
public class WindowWaterMakeBoundOutWithTimestamp {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        environment.setParallelism(2);

        /**
         * 乱序流设置 水位策略
         * 延迟 2秒
         */
        WatermarkStrategy<WaterSensor> watermarkStrategy =
                // forBoundedOutOfOrderness 设置延迟时间
                // 水位线生成器
                WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                // 时间戳生成器
                .withTimestampAssigner(
                        (TimestampAssignerSupplier< WaterSensor>) context -> new TimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                        System.out.println("TimestampAssigner---》 " + element + "recordTimestamp ---> " + recordTimestamp);
                        return element.getVc() * 1000;
                    }
                });

        SingleOutputStreamOperator<WaterSensor> streamOperator =
                environment.socketTextStream("127.0.0.1", 7777)
                .map(new WaterSensorFunction())
                .assignTimestampsAndWatermarks(watermarkStrategy);

        streamOperator
                .keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<WaterSensor, String, String, TimeWindow>.Context context,
                                        Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        String startTime = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss");
                        String endTime = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss");
                        out.collect(
                                "窗口开始时间：" + startTime +
                                "窗口结束时间：" + endTime +
                                " 窗口内数据条数：" + elements.spliterator().estimateSize()
                                + " 窗口内数据：" + elements
                        );
                    }
                })
                .print();

        environment.execute();
    }

}
